package com.shujia.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo2Map {
  def main(args: Array[String]): Unit = {


    val conf: SparkConf = new SparkConf()
      .setAppName("map")
      .setMaster("local")

    //spark  上下文对象
    val sc = new SparkContext(conf)


    /**
      * map : 一行一行处理rdd中的数据
      */

    /**
      * 构建rdd的方法
      * 1、读取文件
      * 2、基于scala集合构建rdd
      *
      */

    //基于scala集合构建rdd
    val listRDD: RDD[Int] = sc.parallelize(List(1, 2, 3, 4, 5, 6, 7, 8, 9), 4)

    println("listRDD分区数据：" + listRDD.getNumPartitions)

    val mapRDD: RDD[Int] = listRDD.map(i => {
      i * 2
    })

    //打印rdd中的数据
    mapRDD.foreach(println)

    /**
      * mapValues ; 处理kv格式rdd的value
      */

    //转换成kv格式
    val kvRDD: RDD[(Int, Int)] = listRDD.map(i => (i, i))

    val mapValuesRDD: RDD[(Int, Int)] = kvRDD.mapValues(i => i * 2)

    mapValuesRDD.foreach(println)


    /**
      * mapPartitions: 一次处理一个分区的数据，返回值需要是一个迭代器
      * mapPartitionsWithIndex: 多了一个下标
      */
    val mapPartitionsRDD: RDD[Int] = listRDD.mapPartitions((iter: Iterator[Int]) => {
      val list: List[Int] = iter.toList
      list.map(i => i * 2).toIterator
    })

    mapPartitionsRDD.foreach(println)


    val mapPartitionsWithIndexRDD: RDD[Int] = listRDD.mapPartitionsWithIndex {
      case (index: Int, iter: Iterator[Int]) =>
        println("当前分区编号：" + index)
        iter
    }

    mapPartitionsWithIndexRDD.foreach(println)

  }

}
